package com.leinao.job.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.leinao.commons.Response;
import com.leinao.commons.ResponseMessage;
import com.leinao.config.EnvironmentConfig;
import com.leinao.config.RedisCacheConfig;
import com.leinao.constant.*;
import com.leinao.job.dao.PaiProjectJobMapper;
import com.leinao.job.dto.PaiProjectJobDto;
import com.leinao.job.model.AiOrderUserCalculationPower;
import com.leinao.job.model.PaiProjectJob;
import com.leinao.job.model.PaiTeamTaskRelation;
import com.leinao.job.service.JobService;
import com.leinao.login.dto.LoginResultDto;
import com.leinao.model.dao.ModelMapper;
import com.leinao.model.model.Model;
import com.leinao.project.dao.PaiDataSetMapper;
import com.leinao.project.dao.ProjectMapper;
import com.leinao.project.model.PaiDataSet;
import com.leinao.project.model.Project;
import com.leinao.project.service.ProjectService;
import com.leinao.task.TaskService;
import com.leinao.util.random.RandomPrimaryCode;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.client.HttpServerErrorException;
import org.springframework.web.client.RestTemplate;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.math.BigDecimal;
import java.util.*;

/**
 * @author  wangshoufa 
 * @date 2018年11月29日 下午3:27:23
 *
 */
@Service
public class JobServiceImpl implements JobService {
	private static Logger logger = LoggerFactory.getLogger(JobServiceImpl.class);
	
	@Autowired
	private RestTemplate restTemplate;

	@Autowired
	private EnvironmentConfig env;
	
	@Autowired
    private TaskService taskService;
	
	@Autowired
	private PaiProjectJobMapper paiProjectJobMapper;
	
	@Autowired
	private ProjectMapper projectMapper;

	@Autowired
	private PaiDataSetMapper paiDataSetMapper;

	@Autowired
	private ModelMapper modelMapper;

	@Autowired
	private ProjectService projectService;

    @Autowired
    RedisCacheConfig redisCacheConfig;

    /**
     * 功能描述：任务操作
     * @param jobId 任务ID
     * @param operateType 1：查看，2：停止或者取消
     * @param token 用户token
     * @return com.leinao.commons.Response<java.lang.Object>
     * @Author: guoliangbo
     * @Date: 2019/5/7 10:14
     */
	@Override
	public Response<Object> jobOperation(Long jobId, Integer operateType, String token) {
		PaiProjectJobDto paiProjectJobDto = paiProjectJobMapper.getProjectJobInfoByJobId(jobId);
		if(ProjectJobHandleEnum.JOB_TYPE_VIEW.getSeq().equals(operateType)) {
			//查看某个任务
			return loadJobDetail(paiProjectJobDto, token);
		} else if(ProjectJobHandleEnum.JOB_TYPE.getSeq().equals(operateType)) {
			return stopJob(paiProjectJobDto.getJobName(), token);
		}
		return null;
	}
	/**
	 * @description 查看任务详情
	 * @param paiProjectJobDto 任务信息
	 * @return
	 * @author  cuidongdong
	 * @date 2019.5.9 11:40:00
	 */
	private Response<Object> loadJobDetail(PaiProjectJobDto paiProjectJobDto, String token) {
		HttpHeaders headers = new HttpHeaders();
		MediaType type = MediaType.parseMediaType("application/json; charset=UTF-8");
		headers.setContentType(type);
		headers.add("Accept", MediaType.APPLICATION_JSON.toString());
		headers.add("Authorization", "Bearer " + token);
		HttpEntity<String> formEntity = new HttpEntity<String>(null, headers);
		String body = "";
		try {
			String config = paiProjectJobDto.getConfig();
			Map<Object,Object> map = new HashMap<Object,Object>();
			if(StringUtils.isNotBlank(config)) {
				map.put("jobConfig", config);
			}
			ResponseEntity<String> responseEntity = restTemplate.exchange(env.getRestServerUrl() + Constant.REST_SERVER_LOAD_JOB_API + paiProjectJobDto.getJobName(),
					HttpMethod.GET, formEntity, String.class);
			logger.debug(String.format("request getJobDetail return: {}", JSON.toJSON(responseEntity)));
			if (responseEntity.getStatusCodeValue() == Constant.RESPONSE_STATUS_SUCCESS) {
				logger.debug("loadJobDetail success");
				body = responseEntity.getBody();
				JSONObject jsonObj = JSON.parseObject(body);
				logger.info(String.format("loadJobDetail resp: {}", jsonObj));
				map.put("jobDetail", jsonObj);
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.GET_SUCCESS.getSeq(), ProjectJobHandleEnum.GET_SUCCESS.getDesc(), map);
			}
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.GET_ERROR.getSeq(), ProjectJobHandleEnum.GET_ERROR.getDesc(), null);
		} catch (Exception e) {
			logger.error("loadJobDetail error");
			logger.error(e.getMessage(), e);
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.GET_ERROR.getSeq(), ProjectJobHandleEnum.GET_ERROR.getDesc(), null);
		}
	}
	/**
	 * @description 停止任务
	 * @param jobName 任务名称
	 * @param token
	 * @return
	 * @author  cuidongdong
	 * @date 2019.5.9 11:40:00
	 */
	private Response<Object> stopJob(String jobName, String token) {
		HttpHeaders headers = new HttpHeaders();
		MediaType type = MediaType.parseMediaType("application/json; charset=UTF-8");
		headers.setContentType(type);
		headers.add("Accept", MediaType.APPLICATION_JSON.toString());
		headers.add("Authorization", "Bearer " + token);
		String json = String.format("{ \"%s\" : \"%s\" }", "value", "STOP");
		logger.info("stopJob req data: {}", json);
		HttpEntity<String> formEntity = new HttpEntity<String>(json, headers);
		try {
			ResponseEntity<String> responseEntity = restTemplate.exchange(env.getRestServerUrl() + Constant.REST_SERVER_STOP_JOB_API.replace("jobName", jobName),
					HttpMethod.PUT, formEntity, String.class);
			logger.info("request stop job return: {}", JSON.toJSON(responseEntity));
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.STOP_SUCCESS.getSeq(), ProjectJobHandleEnum.STOP_SUCCESS.getDesc(), null);
		} catch (Exception e) {
			logger.error("stopJob error");
			logger.error(e.getMessage(), e);
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.STOP_ERROR.getSeq(), ProjectJobHandleEnum.STOP_ERROR.getDesc(), null);
		}
	}


    /**
     * 功能描述：创建任务
     * @param job 创建任务
     * @param userInfo 用户信息
     * @Author: guoliangbo
     * @Date: 2019/5/7 10:15
     */
	@Transactional(rollbackFor=Exception.class)
	@Override
	public synchronized Response<String> createJob(PaiProjectJob job, LoginResultDto userInfo) {

		// 1.0 调试任务类型处理
		if (ProjectJobHandleEnum.JOB_TYPE.getSeq().equals(job.getJobType())) {
			// 同一项目5秒内只能创建一个调试任务
			boolean flag = safetyControl(job.getProjectId());
			if (!flag) {
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.CREATE_JOB_SAFE_CONTROL.getSeq(), ProjectJobHandleEnum.CREATE_JOB_SAFE_CONTROL.getDesc(), null);
			}
		}
		// 任务配置转换为 JSONObject
		JSONObject pobj = JSON.parseObject(job.getConfig());
		// 任务名称不能为空
		if(!pobj.containsKey("jobName") || StringUtils.isBlank(pobj.getString("jobName"))) {
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getSeq(), ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getDesc(), null);
		}
		// apiVersion不能为空
		if(!pobj.containsKey("apiVersion") || StringUtils.isBlank(pobj.getString("apiVersion"))) {
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getSeq(), ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getDesc(), null);
		}
		// 数据集必须有
		if(!pobj.containsKey("datasets")) {
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getSeq(), ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getDesc(), null);
		}
		// 代码必须有
		if(!pobj.containsKey("code")) {
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getSeq(), ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getDesc(), null);
		}
		// 模型必须有
		if(!pobj.containsKey("model")) {
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getSeq(), ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getDesc(), null);
		}

		// 任务名称
		String jobName = pobj.getString("jobName");
		// 验证用户剩余算力是否大于0，只有大于0才能提交任务
        BigDecimal surplus = getUserOrTeamSurplusCalculationPower(jobName,userInfo.getUserId());
        if(surplus.compareTo(BigDecimal.ZERO) <= 0){
            return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.USERS_ARE_UNDERPOWERED.getSeq(), ProjectJobHandleEnum.USERS_ARE_UNDERPOWERED.getDesc(), null);
        }

		try {
			job.setJobName(jobName);
			long currentTimeMillis = System.currentTimeMillis();
			job.setCreateTime(currentTimeMillis);
			job.setUpdateTime(currentTimeMillis);
			Project project = projectMapper.getProject(job.getProjectId());
			Response<Void> response = setJobCodeNo(job, project, userInfo);
			if(response.getMessage().getCode() != ResponseMessage.SUCCESS_CODE) {
				return Response.error(Constant.RESPONSE_STATUS_SUCCESS, response.getMessage().getCode(), response.getMessage().getMessage(), null);
			}
			return taskService.syncRunningJob(userInfo, job.getConfig());
		}
		catch (Exception e) {
			TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
			logger.error("createJob error");
			logger.error(e.getMessage(), e);
			if(e.toString().contains("ResourceAccessException") || e.toString().contains("SocketTimeoutException")) {
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.RUN_ERROR.getSeq(), ProjectJobHandleEnum.RUN_ERROR.getDesc(), String.format("jobName:%s, msg:%s", jobName, e.getMessage()));
			}
			if (e.toString().contains("HttpClientErrorException")) {
			    if (((HttpClientErrorException) e).getStatusCode().value() == Constant.RESPONSE_STATUS_BAD_REQUEST_ERROR) {
				    return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getSeq(), ProjectJobHandleEnum.CREATE_JOB_ARGS_ERROR.getDesc(), String.format("jobName:%s, msg:%s", jobName, e.getMessage()));
			    }
			    if (((HttpClientErrorException) e).getStatusCode().value() == Constant.RESPONSE_STATUS_UNAUTHORIZED_ERROR) {
			    	String responseBody = ((HttpClientErrorException) e).getResponseBodyAsString();
					JSONObject jsonObj = JSON.parseObject(responseBody);
					return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JOB_UNAUTHORIZED.getSeq(), jsonObj.getString("message"), JSON.toJSONString(jsonObj));
			    }
			    if (((HttpClientErrorException) e).getStatusCode().value() ==Constant.RESPONSE_STATUS_INVALID_LINK) {
				    return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.URL_ERROR.getSeq(), ProjectJobHandleEnum.URL_ERROR.getDesc(), String.format("jobName:%s, msg:%s", jobName, e.getMessage()));
			    }
			    if (((HttpClientErrorException) e).getStatusCode().value() == Constant.RESPONSE_STATUS_CONFLICT_ERROR) {
    				logger.warn("jobName {} is conflict", jobName);
    				String name = jobName.substring(0, jobName.length() - 7);
    				String primaryCode = RandomPrimaryCode.getPrimaryCode(6);
    				jobName = name + "-" + primaryCode;
    				pobj.put("jobName", jobName);
    				String config = pobj.toJSONString();
    				return taskService.syncRunningJob(userInfo, config);
			    }
			}
			if (e.toString().contains("HttpServerErrorException")) {
				String responseBody = ((HttpServerErrorException) e).getResponseBodyAsString();
				JSONObject jsonObj = JSON.parseObject(responseBody);
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.HTTP_SERVER_EXCEPTION.getSeq(), jsonObj.getString("message"),JSON.toJSONString(jsonObj));
			}
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.RUN_ERROR.getSeq(), ProjectJobHandleEnum.RUN_ERROR.getDesc(), String.format("jobName:%s, msg:%s", jobName, e.getMessage()));
		}
	}

	/**
	  * 功能描述：查询用户剩余算力
	  * @param jobName 任务名
	  * @param userId 用户id
	  * @return java.math.BigDecimal
	  * @Author: guoliangbo
	  * @Date: 2019/9/27 14:07
	  */
    private BigDecimal getUserOrTeamSurplusCalculationPower(String jobName, String userId) {
        BigDecimal surplus =  BigDecimal.ZERO;
        // 查询一下，是否使用团队算力
        PaiTeamTaskRelation relation = projectMapper.queryTeamTaskRelationByJobName(jobName);
        // 任务是使用团队算力的，故查询团长的算力情况
        if(null != relation){
            userId = relation.getTeamCreaterId();
        }
        // 查询用户剩余算力
        AiOrderUserCalculationPower power = projectMapper.queryAiOrderUserCalculationPower(userId);
        if(null != power){
            surplus = power.getSurplusCalculationPower();
            Jedis jedis = null;
            try {
                JedisPool jedisPool = redisCacheConfig.getJedisPool();
                jedis = jedisPool.getResource();
                List<String> jobs = jedis.hvals(Constant.USER_RUNNING_JOB + userId);
                if( null != jobs && jobs.size() > 0){
                    BigDecimal deductionValue = BigDecimal.ZERO;
                    for(String str : jobs) {
                        deductionValue = deductionValue.add(new BigDecimal(str));
                    }
                    surplus = surplus.subtract(deductionValue);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                if (jedis!=null){
                    jedis.close();
                }
            }
        }
        return surplus;
    }

    /**
	 * @description 保存任务并且获取id
	 * @param job 任务信息
	 * @param job 项目信息
	 * @param userInfo 用户信息
	 * @author cuidongdong
	 * @date 2019.5.9 11:32:00
	 */
	private synchronized Response<Void> setJobCodeNo(PaiProjectJob job, Project project, LoginResultDto userInfo) {
		Integer codeNo = 1;
		List<Integer> coll = projectMapper.getMaxCodeNo(job.getProjectId());
		if(CollectionUtils.isNotEmpty(coll) && coll.size() > 0) {
			codeNo = coll.get(0) + 1;
		}

		// 提交代码分支，并且把分支保存进入任务表git_tag字段
		MultiValueMap<String, String> paramMap = new LinkedMultiValueMap<>();
		paramMap.add("projectId", job.getProjectId().toString());
		paramMap.add("tag", codeNo.toString());
		Response<String> response = projectService.requestFileCenter(paramMap, userInfo, Constant.FILE_CENTER_JOB_INIT_TAG_API);
		if(response.getMessage().getCode() == ResponseMessage.SUCCESS_CODE) {
			job.setGitTag(response.getData());
		} else {
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, response.getMessage().getCode(), response.getMessage().getMessage(), null);
		}

		job.setCodeNo(codeNo);

		// TODO: 此处需要特别注意：该config转换专门为了提供给restServer使用，为了处理数据集给rest-server的是：/data/${username}/${dataSetName}
		// TODO: 2.0入库暂时还是存/data/${userid}/${dataSetName}，以后统一都需要改成/data/${username}/${dataSetName}，并且前端用时datasetId作为标识
		String restServerConfig = null;

		if(StringUtils.isNotBlank(job.getConfig())) {
			JSONObject pobj = JSON.parseObject(job.getConfig());
			String config = job.getConfig();
			if(config.contains("\"minSucceededTaskCount\":\"\"")) {
				config.replace("\"minSucceededTaskCount\":\"\"", "\"minSucceededTaskCount\":null");
			}
			if(config.contains("\"minFailedTaskCount\":\"\"")) {
				config.replace("\"minFailedTaskCount\":\"\"", "\"minFailedTaskCount\":null");
			}

			// outputs处理
			Map<String, Object> outputMap = new HashMap<>();
			outputMap.put("jobId", codeNo);
			pobj.put("output", outputMap);

			// code处理
			pobj.getJSONObject("code").put("timeStamp", project.getGitTimestamp());
			pobj.getJSONObject("code").put("version", job.getGitTag());

			// 在config被修改前复制
			restServerConfig = pobj.toJSONString();

			// model处理
			initModelConfigInfo(job.getProjectId(), pobj, false, userInfo.getUserId());

			// datasets处理
			initDatasetConfigInfo(job.getProjectId(), pobj, false, userInfo.getUserId());

			// TODO: 需要提供任务版本，供底层验证
			config = pobj.toJSONString();
			job.setConfig(config);
		}
		int jobNum = paiProjectJobMapper.createProjectJobRel(job);
		logger.info("createProjectJobRel insert into :{}", jobNum);
		//update project config
		if (job.getJobType() == 1) {
		    Project dbProject = new Project();
			dbProject.setId(job.getProjectId());
			dbProject.setConfig(job.getConfig());
		    projectMapper.editProject(dbProject);
		    logger.debug("editProject config :{}", job.getConfig());
		}

		// 后面的步骤是对job.config设置的值是给rest-server的
		if(StringUtils.isNotBlank(restServerConfig)) {
			JSONObject pobj = JSON.parseObject(restServerConfig);

			// model处理
			initModelConfigInfo(job.getProjectId(), pobj, true, userInfo.getUserId());

			// datasets处理
			initDatasetConfigInfo(job.getProjectId(), pobj, true, userInfo.getUserId());

			// TODO: 需要提供任务版本，供底层验证
			restServerConfig = pobj.toJSONString();
			job.setConfig(restServerConfig);
			logger.debug("editProject for rest-server config :{}", job.getConfig());
		}

		return Response.ok();
	}

	/**
	 * 获取用户调试任务的JupyterURL
	 * @param projectId 项目id
	 * @param token 用户token
	 * @return 结果
	 * @author wangshoufa 
	 * @date 2019年3月22日
	 *
	 */
	@Override
	public Response<String> jobJupyterUrl(Integer projectId, String token) {
		List<String> list = Lists.newArrayList();
		try {
			list = paiProjectJobMapper.getDebugJob(projectId);
			if(CollectionUtils.isEmpty(list)) {
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.NO_JOB.getSeq(), "暂无调试任务", JSON.toJSONString(list));
			}
			for(int i=0; i< list.size(); i++) {
				String jobName = list.get(i);
				ResponseEntity<String> result = restTemplate.getForEntity(env.getFrameworksUrl() + jobName, String.class);
				if(result.getStatusCode().OK.value() ==Constant.RESPONSE_STATUS_SUCCESS) {
					logger.debug("req get jobDetail resp body:" + JSON.toJSON(result.getBody()));
					JSONObject reponse = JSON.parseObject(result.getBody());
					Object object = reponse.get("aggregatedFrameworkStatus");
					Object frameworkStatus = JSON.parseObject(object.toString()).get("frameworkStatus");
					String frameworkState = (String) JSON.parseObject(frameworkStatus.toString()).get("frameworkState");
					Object summarizedFrameworkInfo = reponse.get("summarizedFrameworkInfo");
					String executionType = (String) JSON.parseObject(summarizedFrameworkInfo.toString()).get("executionType");
					if((executionType.equals("START") && frameworkState.contains("WAITING")) ||
							(executionType.equals("START") && frameworkState.contains("RUNNING"))||
							(executionType.equals("START") && frameworkState.contains("LAUNCHED"))||
							(executionType.equals("START") && frameworkState.contains("APPLICATION_COMPLETED"))) {
						logger.info("waiting or running jobName: {}", jobName);
						Object aggregatedTaskRoleStatuses = JSON.parseObject(object.toString()).get("aggregatedTaskRoleStatuses");
						JSONObject array = JSON.parseObject(aggregatedTaskRoleStatuses.toString());
						if(array!=null && array.size() > 0) {
							Set<String> keys = array.keySet();
							Iterator iterator = keys.iterator();
							while(iterator.hasNext()){
								String key = (String) iterator.next();
								Object taskRole = array.get(key);
								Object taskStatuses = JSON.parseObject(taskRole.toString()).get("taskStatuses");
								JSONArray taskStatusArray = JSON.parseObject(taskStatuses.toString()).getJSONArray("taskStatusArray");
								if(!(JSON.parseObject(taskStatusArray.get(0).toString()).containsKey("containerLogHttpAddress"))) {
									return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), null);
								}
								String containerLog = (String) JSON.parseObject(taskStatusArray.get(0).toString()).get("containerLogHttpAddress");
								if(StringUtils.isBlank(containerLog)) {
									return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), null);
								}
								String[] ipInfo = containerLog.split("/node");
								String ip = ipInfo[0];
								logger.info("containerLog: {}", containerLog);
								String stdoutLog = restTemplate.getForObject(containerLog, String.class);
								logger.debug("stdoutLog: {}", JSON.toJSON(stdoutLog));
								String stdoutUrl = new StringBuffer(ip).append(getUrl(stdoutLog)).toString();
								if(stdoutUrl.contains("?start=")) {
									stdoutUrl = stdoutUrl.replace("?start=-4096", "?start=0");
								}
								//stdoutUrl日志过长需要处理
								String outLog = restTemplate.getForObject(stdoutUrl, String.class);
								logger.debug("last outLog: {}", JSON.toJSON(outLog));
								String jupyterUrl = "";
								if(StringUtils.isNotBlank(outLog) && outLog.contains("/?token=")) {
									if(outLog.contains("referer=")) {
										String[] url = outLog.split("referer=");
										url = url[1].split("\n");
										jupyterUrl = url[0];
									}
									else {
										ResponseEntity<String> resp = restTemplate.getForEntity(env.getRestServerUrl() + Constant.REST_SERVER_LOAD_JOB_API + jobName + "/ssh", String.class);
										if(resp.getStatusCode().OK.value() == Constant.RESPONSE_STATUS_SUCCESS) {
											logger.debug("req get job SSH Info resp body:" + JSON.toJSON(resp.getBody()));
											JSONObject body = JSON.parseObject(resp.getBody());
											if(body.containsKey("containers")) {
												JSONArray ssh_array = JSON.parseObject(body.toString()).getJSONArray("containers");
												String sshIp = (String) JSON.parseObject(ssh_array.get(0).toString()).get("sshIp");
												Object obj = JSON.parseObject(ssh_array.get(0).toString()).get("sshPort");
												String sshPort = String.valueOf(obj);
												if(StringUtils.isNotBlank(sshIp) && sshPort.compareTo("0") > 0) {
													jupyterUrl = new StringBuffer(sshIp).append(":").append(sshPort).append("/lab").toString();
												}
											}
											else {
												return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), String.format("jobName: %s", jobName));
											}
										}
										else {
											return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), String.format("jobName: %s", jobName));
										}
									}
									String[] Token = outLog.split("/?token=");
									String[] jupyterToken = Token[1].split("\n");
									jupyterUrl = new StringBuffer(jupyterUrl).append("?token=").append(jupyterToken[0]).toString();
									String jupyterURL = jupyterUrl.replaceAll("\\r", "");
									logger.info("get jupyterURL success: {}", jupyterURL);
									return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_SUCCESS.getSeq(), ProjectJobHandleEnum.JUPYTER_SUCCESS.getDesc(), jupyterURL);
								}
								else {
									return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), containerLog);
								}
							}
						}
						else {
							return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), result.getBody());
						}
					}
					else {
						if (i == list.size() - 1) {
							return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_NO_RUN_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_NO_RUN_ERROR.getDesc(), String.format("jobName: %s", jobName));
						} else {
							continue;
						}
					}
				}
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_JOB_INFO_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_JOB_INFO_ERROR.getDesc(), JSON.toJSONString(JSON.toJSON(result)));
			}
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_NO_RUN_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_NO_RUN_ERROR.getDesc(), null);
		} catch (Exception e) {
			logger.error("get jobJupyterUrl error");
			logger.error(e.getMessage(), e);
			if(e.toString().contains("HttpServerErrorException") || e.toString().contains("HttpClientErrorException")) {
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), e.getMessage());
			}
			if(e.toString().contains("ResourceAccessException") || e.toString().contains("ConnectException")) {
				return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), e.getMessage());
			}
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getSeq(), ProjectJobHandleEnum.JUPYTER_GET_URL_ERROR.getDesc(), e.getMessage());
		}
	}

	/**
	 * @description  url处理公共方法
	 * @param url
	 * @return
	 * @author wangshoufa
	 * @date 2019年3月13日
	 */
	public static String getUrl(String url) {
		try {
			Document doc = Jsoup.parse(url);
			Elements imgs = doc.getElementsByTag("a");
			for (Element el : imgs) {
			    String nodeUrl = el.attr("href");
			    if (nodeUrl.contains("stdout")) {
			        return nodeUrl;
			    }
			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
		return null;
	}
	/**
	 * @description 创建任务控制，5s只能创建一个
	 * @param projectId 项目id
	 * @return boolean true或者false
	 * @author wangshoufa
	 * @date 2019年3月13日
	 */
	private boolean safetyControl(Long projectId) {
		// 5s只能新建一个
		Integer sendTimesSecond = paiProjectJobMapper.queryProjectDebugJobFrequency(projectId, 0);
		if (sendTimesSecond > 0) {
			logger.debug("in 5 seconds not allowed to create debugJob for 5 seconds already have a debugJob: {}", sendTimesSecond);
			return false;
		}
		return true;
	}

	/**
	 * 功能描述：编辑任务备注
	 * @param userId 用户token
	 * @param projectId 项目id
	 * @param jobName 任务名称
	 * @param remark 任务备注
	 * @return com.leinao.commons.Response<java.lang.String>  返回编辑结果
	 * @Author: guoliangbo
	 * @Date: 2019/5/17 16:49
	 */
	@Override
	public Response<String> jobRemarkEdit(Integer projectId, String jobName, String remark, String userId) {
		//1.验证项目任务是否属于该用户，如不属于，无权操作
		PaiProjectJob projectJob = paiProjectJobMapper.selectProjectJob(projectId, jobName, userId);
		if( null == projectJob ){
			return  Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.NOT_JOB_OWNER.getSeq(), ProjectJobHandleEnum.NOT_JOB_OWNER.getDesc(), "非任务拥有者！");
		}
		//2.如任务属于该用户，修改任务备注
		projectJob.setRemark(remark);
		//更新任务信息
		Integer num = paiProjectJobMapper.updateJobInfo(projectJob);
		if(num > 0 ){
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.UPDATE_SUCCESS.getSeq(), ProjectJobHandleEnum.UPDATE_SUCCESS.getDesc(), "更新成功！");
		}else{
			return Response.ok(Constant.RESPONSE_STATUS_SUCCESS, ProjectJobHandleEnum.UPDATE_FAILED.getSeq(), ProjectJobHandleEnum.UPDATE_FAILED.getDesc(), "更新失败！");
		}
	}

	/**
	 * @description 初始化任务配置中的数据集信息
	 * @param projectId 项目ID
	 * @param pobj 任务config配置
	 * @param saveForRestServer 是否为了rest-server提供修改
	 * @author lizi
	 * @date 2019.6.4 20:32:00
	 */
	private void initDatasetConfigInfo(Long projectId, JSONObject pobj, boolean saveForRestServer, String userId) {
		List<PaiDataSet> dataSetList = paiDataSetMapper.getDatasetListByProjectId(projectId);
		dataSetList = dataSetList == null ? Collections.emptyList() : dataSetList;

		// 获取所有进行分享的数据集ID集合
		List<Long> shareDataSetIdList = paiDataSetMapper.getShareDataSetIdList(userId);
		shareDataSetIdList = shareDataSetIdList == null ? Collections.emptyList() : shareDataSetIdList;

		// 数据集信息初始化
		List<Map<String, Object>> modleMapList = new ArrayList<>();
		for(PaiDataSet paiDataSet : dataSetList) {
			Map<String, Object> datasetMap = new LinkedHashMap<>();
			datasetMap.put("datasetId", paiDataSet.getName());
			datasetMap.put("owner", paiDataSet.getUserId());
			String dataSetPath = paiDataSet.getPath();
			dataSetPath = dataSetPath.replaceFirst(Constant.DATASET_PATH, Constant.DATASET__CONTAINER_PATH);
			datasetMap.put("mountPath", dataSetPath);

			// 对于rest-server需要把路径中的userid替换成account
			if(saveForRestServer && StringUtils.isNotEmpty(paiDataSet.getUserId())) {
				dataSetPath = dataSetPath.replaceFirst(paiDataSet.getUserId(), paiDataSet.getAccount());
				datasetMap.put("mountPath", dataSetPath);
			}

			// 数据集是否共享
			boolean share = DatasetTypeEnum.PUBLIC.getType().equals(paiDataSet.getType()) || shareDataSetIdList.contains(paiDataSet.getId());
			datasetMap.put("share", share);
			datasetMap.put("token", "");
			modleMapList.add(datasetMap);
		}

		pobj.put("datasets", modleMapList);
	}

	/**
	 * @description 初始化任务配置中的模型信息
	 * @param projectId 项目ID
	 * @param pobj 任务config配置
	 * @author lizi
	 * @date 2019.6.4 20:32:00
	 */
	private void initModelConfigInfo(Long projectId, JSONObject pobj, boolean saveForRestServer, String userId) {
		List<Model> modelList = modelMapper.getModelListByProjectId(projectId);
		modelList = modelList == null ? Collections.emptyList() : modelList;

		// 获取所有进行分享的模型ID集合
		List<Long> shareModelIdList = modelMapper.getModelSetIdList(userId);
		shareModelIdList = shareModelIdList == null ? Collections.emptyList() : shareModelIdList;

		// 数据集信息初始化
		List<Map<String, Object>> modelMapList = new ArrayList<>();
		for(Model model : modelList) {
			Map<String, Object> modelMap = new LinkedHashMap<>();
			modelMap.put("modelId", model.getModelName());
			modelMap.put("owner", model.getUserId());
			modelMap.put("mountPath", Constant.MODELS_CONTAINER_PATH + Constant.FILE_PATH_SEPARATOR + model.getAccount() +  Constant.FILE_PATH_SEPARATOR + model.getModelName());

			if(!saveForRestServer) {
				modelMap.put("modelDataId", model.getId());
			}

			// 数据集是否共享
			boolean share = ModelTypeEnum.PUBLIC.getType().equals(model.getType()) || shareModelIdList.contains(model.getId());
			modelMap.put("share", share);
			modelMap.put("token", "");
			modelMapList.add(modelMap);
		}

		pobj.put("models", modelMapList);
		// 删除之前老数据
		pobj.remove("model");
	}
}
